package com.jiuzhou.crawler.base.HdfsDao;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
	 * 项目名称：大数据爬虫
	 * @author yanganzhen
	 * version-code：1.0
     * 类说明：HDFS文件操作Dao层实现类
     * 创建时间：2016年12月23日
	 */
     public class HdfsUtils{
    		static {
    			String hadoopHome = System.getenv("HADOOP_HOME");
    			if (hadoopHome != null && !hadoopHome.isEmpty())
    				System.setProperty("hadoop.home.dir", hadoopHome);
    		}

    		/**方法名称：WriteToHdfs()
    		 * 方法说明：写入到hdfs
    		 * @param filePathName 
    		 * @param content void
    		 */
			public synchronized static void WriteToHdfs(String filePathName, String content) {
				FileSystem fs = null;
				FSDataOutputStream out = null;
				try {
					fs = FileSystem.get(URI.create(filePathName), new Configuration());
					out = fs.create(new Path(filePathName));
					out.write(content.getBytes("UTF-8"));
					// or using ----out.writeBytes(content);
				} catch (IOException e) {
					e.printStackTrace();
				} finally {
					if (out != null) {
						try {
							out.close();
							out = null;
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
					if (fs != null) {
						try {
							fs.close();
							fs = null;
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
				}
			}

			/**方法名称：WriteToHdfsFromStream()
			 * 方法说明：从一个文件写入到另一个文件
			 * @param fromFile
			 * @param toFile void
			 */
			public synchronized static void WriteToHdfsFromStream(String fromFile, String toFile) {
				// TODO Auto-generated method stub
				
			}

			/**方法名称：ReadFromHdfs()
			 * 方法说明：从hdfs读取文件内容
			 * @param filePathName
			 * @return String
			 */
			public  synchronized static String ReadFromHdfs(String filePathName) {
				FileSystem fs = null;
				FSDataInputStream in = null;
				try{
					fs = FileSystem.get(URI.create(filePathName), new Configuration());
					Path path = new Path(filePathName);
					in = fs.open(path);
					// IOUtils.copyBytes(in, System.out,
					// fs.getFileStatus(path).getLen(),true);
					byte[] b = new byte[(int) fs.getFileStatus(path).getLen()];
					in.readFully(b);
					return new String(b);
				} catch (IOException e) {
					e.printStackTrace();
					return "";
				} finally {
					if (in != null) {
						try {
							in.close();
							in = null;
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
					if (fs != null) {
						try {
							fs.close();
							fs = null;
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
				}
			}
			
			/**方法名称：getFilesUnderFolder()
			 * 方法说明：获取路径下所有的文件名称
			 * @param folderPath
			 * @param pattern
			 * @return List<Path>
			 */
			public synchronized static  List<Path> getFilesUnderFolder(Path folderPath, String pattern) throws IOException {  
				FileSystem fs = null;
				fs = FileSystem.get(URI.create(folderPath.toString()), new Configuration());
				  List<Path> paths = new ArrayList<Path>();  
				  if (fs.exists(folderPath)) {  
				   FileStatus[] fileStatus = fs.listStatus(folderPath);  
				   for (int i = 0; i < fileStatus.length; i++) {  
				    FileStatus fileStatu = fileStatus[i];  
				   // if (!fileStatu.isDir()) {//只要文件  
				     Path oneFilePath = fileStatu.getPath();  
				     if (pattern == null && !oneFilePath.getName().contains("detail")) {  
				      paths.add(oneFilePath);  
				     } else {  
				      if (oneFilePath.getName().contains(pattern) && !oneFilePath.getName().contains("detail")) {  
				       paths.add(oneFilePath);  
				      }  
				     }    
				  
				  //  }  
				   }  
				  }  
				  return paths;  
				 }  
            
			
			/**方法名称：readByLineParamName()
			 * 方法说明：根据行变量名获取文件名称
			 * @param filePath
			 * @param paramName
			 * @return List<String>
			 */
			public synchronized static List<String> readByLineParamName(Path filePath,String paramName) throws IOException{
				FileSystem fs = null;
				FSDataInputStream in = null;
				BufferedReader br =null;  
				List<String> allIdLines =new ArrayList<String>();
				String line = "";
					fs = FileSystem.get(URI.create(filePath.toString()), new Configuration());
					in = fs.open(filePath);
				    br = new BufferedReader(new InputStreamReader(in));  
				       while ((line = br.readLine()) != null) {  
				    	  if(line.contains(paramName)){
				    		  allIdLines.add(line);
				    	  }//end of if
	            }//end of while
					return allIdLines;
		 }//end of method

			/**方法名称：getAllFileByRootPath()
			 * 方法说明：根据根路径
			 * @param rootPath
			 * @return List<List<Path>>
			 */
			public synchronized static  List<List<Path>> getAllFileByRootPath(String rootPath) throws IOException{
				Path toPath = new Path(rootPath);//转化为Path类型
				List<List<Path>> dataFilePaths = new  ArrayList<List<Path>>(); //用list集合存放所有爬取类型的
					 List<Path> typeNamePaths = HdfsUtils.getFilesUnderFolder(toPath, "");
					for(Path root : typeNamePaths){
						dataFilePaths.add(HdfsUtils.getFilesUnderFolder(root, ""));
					}
				return dataFilePaths;		
			}
    		
}
